package main.java.sales

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession


/**
  * ProductsMain
  *
  * @author zhangyimin
  * @date 2018-11-09 上午9:50
  * @version 1.0
  */
object ProductsInfoMain {

  def main(args: Array[String]): Unit = {


    //屏蔽日志
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
    val sparkSession = SparkSession.builder().appName("productsInfo").master("local").getOrCreate()
    val sc = sparkSession.sparkContext
    val salesRDD = sc.textFile("hdfs://10.16.7.36:9000/data/input/sales/sales")
    val prodRDD = sc.textFile("hdfs://10.16.7.36:9000/data/input/sales/products")
    val salesLines = salesRDD.map(_.split(","))
    val salesMaps = salesLines.map(x => {
      new sales(x(0), x(2).substring(0, 4), (x(6).toDouble))
    })
    import sparkSession.implicits._
    val salesDF = salesMaps.toDF()
    salesDF.createOrReplaceTempView("SALES")
    println("=====================华丽丽的分割线======================")
    val prodLines = prodRDD.map(_.split(","))
    val prodMaps = prodLines.map(x => {
      new products(x(0), x(1))
    })
    val prodDF = prodMaps.toDF()
    prodDF.createOrReplaceTempView("PRODUCTS")
    //    分步创建视图
    //    val res1DF=sparkSession.sql(
    //      "SELECT\n     b.prod_name                  name,\n     sum(a.prod_price)           totalAount,\n     a.prod_year    year_id\n   FROM SALES   a, PRODUCTS   b\n   where a.prod_id = b.prod_id\n   group by a.prod_year, b.prod_name"
    //    ).toDF("name","totalAount","year_id")

    //    res1DF.createOrReplaceTempView("PRODUCTS_INFO")
    //
    //    sparkSession.sql(
    //      "select\n  c.name,\n  sum(if(c.year_id = '1998', c.totalAount, 0))  year_1998,\n  sum(if(c.year_id = '1999', c.totalAount, 0))  year_1999,\n  sum(if(c.year_id = '2000', c.totalAount, 0))  year_2000,\n  sum(if(c.year_id = '2001', c.totalAount, 0))  year_2001\nfrom\n  PRODUCTS_INFO   c\ngroup by c.name"
    //    ).show
    //一步到位
    val resDF = sparkSession.sql(
      "select\n  c.name,\n  round(sum(if(c.year_id = '1998', c.totalAount, 0)),2)  year_1998,\n  round(sum(if(c.year_id = '1999', c.totalAount, 0)),2)  year_1999,\n  round(sum(if(c.year_id = '2000', c.totalAount, 0)),2)  year_2000,\n  round(sum(if(c.year_id = '2001', c.totalAount, 0)),2)  year_2001\nfrom\n  (SELECT\n       b.prod_name                  name,\n       sum(a.prod_price)           totalAount,\n       a.prod_year    year_id\n     FROM SALES   a, PRODUCTS   b\n     where a.prod_id = b.prod_id\n     group by a.prod_year, b.prod_name)   c\ngroup by c.name"
    ).show()
    sparkSession.stop();
  }


  case class sales(prod_id: String, prod_year: String, prod_price: Double)

  case class products(prod_id: String, prod_name: String)


}
